package com.atbeijing.bigdata.spark.core.req

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Spark01_Req_HotCategoryTop10_1 {

    def main(args: Array[String]): Unit = {

        val conf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10")
        val sc = new SparkContext(conf)

        // TODO 1. 读取用户行为数据
        val action = sc.textFile("data/user_visit_action.txt")

        // RDD不保存数据，处理一次就会重新执行一次
        action.cache()

        // TODO 2. 统计品类的点击数量
        // 点击数据 => (品类, 1) => (品类, sum)
        val clickDatas = action.filter(
            line => {
                // 拆分数据
                val datas: Array[String] = line.split("_")
                datas(6) != "-1"
            }
        )
        val clickCount = clickDatas.map(
            line => {
                val datas = line.split("_")
                (datas(6), 1)
            }
        ).reduceByKey(_+_)

        // (1, 100)
        // (2, 50) => 2( 50, 30 )
        // (3, 50) => 3( 50, 20)
        // (4, 30)

        // TODO 3. 统计品类的下单数量
        // 下单数据 => (品类1, 1),(品类2, 1),(品类3, 1) => (品类1, sum)(品类2, sum)(品类3, sum)
        val orderDatas = action.filter(
            line => {
                // 拆分数据
                val datas: Array[String] = line.split("_")
                datas(8) != "null"
            }
        )
        val orderCount = orderDatas.flatMap(
            line => {
                val datas = line.split("_")
                val orderids = datas(8).split(",")
                orderids.map((_,1))
            }
        ).reduceByKey(_+_)

        // TODO 4. 统计品类的支付数量
        // 支付数据 => (品类, 1) => (品类, sum)
        val payDatas = action.filter(
            line => {
                // 拆分数据
                val datas: Array[String] = line.split("_")
                datas(10) != "null"
            }
        )
        val payCount = payDatas.flatMap(
            line => {
                val datas = line.split("_")
                val orderids = datas(10).split(",")
                orderids.map((_,1))
            }
        ).reduceByKey(_+_)

        // TODO 5. 将统计的结果进行排序（ 点击数量, 下单数量, 支付数量  ）
        // TODO 6. 将结果取前10条
        //val top10 = clickCount.sortBy(_._2, false).take(10)

        // Tuple => sort
        // (品类， 点击数量)
        // (品类， 下单数量)    => (品类， (点击数量,下单数量,支付数量))
        // (品类， 支付数量)

        // join
        //val value: RDD[(String, ((Int, Int), Int))] = clickCount.join(orderCount).join(payCount)
//        clickCount.join(orderCount)
//        val connect: RDD[(String, (Iterable[Int], Iterable[Int], Iterable[Int]))] = clickCount.cogroup(orderCount, payCount)
//
//        val connectMap = connect.map {
//            case ( cat, ( iterClick, iterOrder, iterPay ) ) => {
//                var clickCnt = 0
//                var orderCnt = 0
//                var payCnt = 0
//                val iterator1: Iterator[Int] = iterClick.iterator
//                if ( iterator1.hasNext ) {
//                    clickCnt = iterator1.next()
//                }
//
//                val iterator2: Iterator[Int] = iterOrder.iterator
//                if ( iterator2.hasNext ) {
//                    orderCnt = iterator2.next()
//                }
//
//                val iterator3: Iterator[Int] = iterPay.iterator
//                if ( iterator3.hasNext ) {
//                    payCnt = iterator3.next()
//                }
//
//                ( cat, (clickCnt, orderCnt, payCnt) )
//            }
//        }
//
//        val top10 = connectMap.sortBy(_._2, false).take(10)

        // (品类， 点击数量) => (品类， (点击数量,0,0))
        // (品类， 下单数量) => (品类， (0,下单数量,0))
        //                => (品类， (点击数量,下单数量,0))
        // (品类， 支付数量) => (品类， (0,0,支付数量))
        //                =>  (品类， (点击数量,下单数量,支付数量))
        val clickTuple = clickCount.map {
            case ( k, v ) => {
                (k, (v, 0, 0))
            }
        }
        val orderTuple = orderCount.map {
            case ( k, v ) => {
                (k, (0, v, 0))
            }
        }
        val payTuple = payCount.map {
            case ( k, v ) => {
                (k, (0, 0, v))
            }
        }

        val top10 = clickTuple.union(orderTuple).union(payTuple).reduceByKey(
            (t1, t2) => {
                (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3)
            }
        ).sortBy(_._2, false).take(10)


        top10.foreach(println)

        sc.stop()

    }
}
